use std::sync::Arc;

#[cfg(feature = "postgres")]
use crate::cache::get_pg_pool;
use crate::cache::{get_mysql_pool, IndexChange};
use crate::sqlx::SqlxDefineTrait;
use crate::{cache, log_db, message_db, run_by_store, sqlx_run_by_store, ROCKSDB};
use rust_rocksdb::WriteBatchWithTransaction;
use visit_dd_core::model::DataStoreEngine;
use visit_dd_core::proj::{VddLogPart, VddMessage, VddUser, APP_CONFIG};
use visit_dd_core::{alog, i18n, Ares, UserId};

pub async fn u_load(uid: &UserId, with_password: bool, with_pri_key: bool) -> VddUser {
    run_by_store!(
        impl_rdb::u_load,
        impl_sqlx::u_load,
        uid,
        with_password,
        with_pri_key
    )
}

pub async fn u_all() -> Vec<Arc<VddUser>> {
    run_by_store!(impl_rdb::u_all, impl_sqlx::u_all)
}

pub async fn u_count_all() -> usize {
    run_by_store!(impl_rdb::u_count_all, impl_sqlx::u_count_all)
}

pub async fn u_register(user: &VddUser) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::u_register(&mut batch, user)
    } else {
        let ds = APP_CONFIG.meta_store.clone().unwrap();
        match ds.store_engine {
            #[cfg(feature = "postgres")]
            DataStoreEngine::Postgres => get_pg_pool(ds.as_ref()).await.u_register(user).await?,
            DataStoreEngine::Mysql => get_mysql_pool(ds.as_ref()).await.u_register(user).await?,
        };
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(VddLogPart::LpAction(
            user.id.to_owned(),
            "注册为用户".into()
        )),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn u_select_by_phone(phone: &str) -> Option<VddUser> {
    run_by_store!(
        impl_rdb::u_select_by_phone,
        impl_sqlx::u_select_by_phone,
        phone
    )
}

pub async fn u_check_phone_unique(phone: &str) -> bool {
    run_by_store!(
        impl_rdb::u_check_phone_unique,
        impl_sqlx::u_check_phone_unique,
        phone
    )
}

pub async fn u_change_name(uid: &UserId, new_name: &str) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::u_change_name(&mut batch, uid, new_name)
    } else {
        sqlx_run_by_store!(impl_sqlx::u_change_name, uid, new_name)?
    }
    cache::update_label_cache(&mut batch, uid.0, new_name.into());
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(VddLogPart::LpAction(uid.clone(), "修改用户名".into())),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn u_change_password(uid: &UserId, old_pwd: &str, new_pwd: &str) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::change_password(&mut batch, uid, old_pwd, new_pwd)?;
    } else {
        sqlx_run_by_store!(impl_sqlx::change_password, uid, old_pwd, new_pwd)?
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(VddLogPart::LpAction(uid.to_owned(), "修改登录密码".into())),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn u_change_role(target_uid: &UserId, is_admin: bool, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    match APP_CONFIG.meta_store.as_ref() {
        None => impl_rdb::u_change_role(&mut batch, target_uid, is_admin),
        Some(..) => sqlx_run_by_store!(impl_sqlx::u_change_role, target_uid, is_admin)?,
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.to_owned(), "修改用户角色".into()),
            VddLogPart::LpUser(target_uid.to_owned()),
        ),
    );
    message_db::save(
        &mut batch,
        &mut index_changes,
        &VddMessage::new(operator, target_uid, i18n!(1010).as_ref()),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?
    }
    ares
}

pub async fn u_change_key(uid: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::change_key(&mut batch, uid);
    } else {
        sqlx_run_by_store!(impl_sqlx::u_change_key, uid)?
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(VddLogPart::LpAction(uid.to_owned(), "刷新密钥对".into())),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn u_after_login(uid: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::u_after_login(&mut batch, uid)
    } else {
        sqlx_run_by_store!(impl_sqlx::u_after_login, uid)?
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(VddLogPart::LpAction(uid.to_owned(), "登录".into())),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

mod impl_rdb {
    use crate::keys::VddKey;
    use crate::{cache, count_in_key_head, load_by_key_head, CF_META, CF_UNIQUE, ROCKSDB};
    use chrono::Local;
    use rust_rocksdb::WriteBatchWithTransaction;
    use serde_json::json;
    use std::sync::Arc;
    use visit_dd_core::i18n_text::{E_ARG_BAD, E_ROCKSDB};
    use visit_dd_core::proj::{VddUser, VddUserRole};
    use visit_dd_core::{i18n, Astr, UserId};

    pub(super) fn u_load(uid: &UserId, with_password: bool, with_pri_key: bool) -> VddUser {
        let key = VddKey::EntityAppUser(uid.clone()).to_string();
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        if let Ok(value) = ROCKSDB.get_cf(cf_meta, key) {
            if let Some(ba) = value {
                let mut user = serde_json::from_slice::<VddUser>(ba.as_slice()).unwrap();
                // update_label_cache(rdb, user.id.0, user.name.clone());
                if !with_password {
                    user.password = Arc::from("");
                }
                if !with_pri_key {
                    user.pri_key = Arc::from("");
                }
                user
            } else {
                panic!("failed to load user {}", uid.0)
            }
        } else {
            panic!("failed to load user {}", uid.0)
        }
    }

    pub(super) fn u_all() -> Vec<Arc<VddUser>> {
        let kh = VddKey::EntityAppUser(UserId(0));
        load_by_key_head(&kh)
            .iter()
            .map(|x| {
                let mut user = serde_json::from_slice::<VddUser>(x).unwrap();
                // update_label_cache(rdb, user.id.0, user.name.clone());
                user.password = Arc::from("");
                user.pri_key = Arc::from("");
                user.into()
            })
            .collect::<Vec<Arc<VddUser>>>()
    }

    pub(super) fn u_count_all() -> usize {
        let key = VddKey::EntityAppUser(UserId(0));
        count_in_key_head(&key, |_| true)
    }

    pub(super) fn u_register(batch: &mut WriteBatchWithTransaction<true>, user: &VddUser) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let v = json!(user).to_string();
        batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(user.id.clone()).to_string(),
            v,
        );
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        batch.put_cf(
            &cf_u,
            VddKey::UniquePhone(user.phone.clone()).to_string(),
            user.id.0.to_le_bytes(),
        );
        cache::update_label_cache(batch, user.id.0, &user.name);
    }

    pub(super) fn u_select_by_phone(phone: &str) -> Option<VddUser> {
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        let k = VddKey::UniquePhone(phone.into()).to_string();
        let mut r: Option<VddUser> = None;
        if let Some(v) = ROCKSDB.get_cf(&cf_u, k).expect(i18n!(E_ROCKSDB).as_ref()) {
            let uid = u64::from_le_bytes(v.as_slice().try_into().expect(i18n!(E_ARG_BAD).as_ref()));
            let user = u_load(&UserId(uid), true, true);
            // update_label_cache(rdb, user.id.0, user.name.clone());
            r = Some(user);
        }
        r
    }

    pub(super) fn u_check_phone_unique(phone: &str) -> bool {
        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
        let key = VddKey::UniquePhone(phone.into()).to_string();
        if let Ok(value) = ROCKSDB.get_cf(&cf_u, &key) {
            if let Some(_) = value {
                false
            } else {
                true
            }
        } else {
            false
        }
    }

    pub(super) fn u_change_name(
        batch: &mut WriteBatchWithTransaction<true>,
        uid: &UserId,
        new_name: &str,
    ) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let mut user = u_load(uid, true, true);
        user.name = new_name.into();
        batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(uid.clone()).to_string(),
            json!(user).to_string(),
        )
    }

    pub(super) fn change_password(
        batch: &mut WriteBatchWithTransaction<true>,
        uid: &UserId,
        old_pwd: &str,
        new_pwd: &str,
    ) -> Result<(), Astr> {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let mut user = u_load(uid, true, true);
        if old_pwd != user.password.as_ref() {
            return Err(i18n!(1009));
        }
        user.password = new_pwd.into();
        Ok(batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(uid.clone()).to_string(),
            json!(user).to_string(),
        ))
    }

    pub(super) fn u_change_role(
        batch: &mut WriteBatchWithTransaction<true>,
        target_uid: &UserId,
        is_admin: bool,
    ) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let mut user = u_load(target_uid, true, true);
        user.role = if is_admin {
            VddUserRole::UrAdmin
        } else {
            VddUserRole::UrUser
        };
        batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(target_uid.clone()).to_string(),
            json!(user).to_string(),
        );
    }

    pub(super) fn u_after_login(batch: &mut WriteBatchWithTransaction<true>, uid: &UserId) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let mut user = u_load(uid, true, true);
        user.last_login_ts = Some(Local::now().naive_local());
        batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(uid.clone()).to_string(),
            json!(user).to_string(),
        );
    }

    pub(super) fn change_key(batch: &mut WriteBatchWithTransaction<true>, uid: &UserId) {
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let mut user = u_load(uid, true, true);
        user.set_key_pair();
        batch.put_cf(
            cf_meta,
            VddKey::EntityAppUser(user.id.clone()).to_string(),
            json!(user).to_string(),
        );
    }
}

mod impl_sqlx {
    use crate::keys::VddKey;
    use crate::sqlx::SqlxDefineTrait;
    use chrono::Local;
    use std::sync::Arc;
    use visit_dd_core::proj::{VddUser, VddUserRole};
    use visit_dd_core::{Ares, UserId};

    pub(super) async fn u_load(
        pool: &impl SqlxDefineTrait,
        uid: &UserId,
        with_password: bool,
        with_pri_key: bool,
    ) -> VddUser {
        let sql_r = pool.load_entity(uid.0).await.unwrap();
        let mut user = serde_json::from_slice::<VddUser>(sql_r.as_bytes()).unwrap();
        if !with_password {
            user.password = Arc::from("");
        }
        if !with_pri_key {
            user.pri_key = Arc::from("");
        }
        user
    }

    pub(super) async fn u_all(pool: &impl SqlxDefineTrait) -> Vec<Arc<VddUser>> {
        let sql_r = pool
            .load_entities(VddKey::EntityAppUser(UserId(0)), false)
            .await
            .unwrap();
        sql_r
            .iter()
            .map(|x| serde_json::from_slice::<VddUser>(x.as_bytes()).unwrap())
            .map(|x| Arc::from(x))
            .collect::<Vec<Arc<VddUser>>>()
    }

    pub(super) async fn u_count_all(pool: &impl SqlxDefineTrait) -> usize {
        pool.count_entities(VddKey::EntityAppUser(UserId(0)))
            .await
            .unwrap()
    }

    pub(super) async fn u_select_by_phone(
        pool: &impl SqlxDefineTrait,
        phone: &str,
    ) -> Option<VddUser> {
        let vk = VddKey::UniquePhone(phone.into());
        let uid = pool.get_unique(vk).await.unwrap();
        let uid = UserId(uid);
        let user = u_load(pool, &uid, true, true).await;
        Some(user)
    }

    pub(super) async fn u_check_phone_unique(pool: &impl SqlxDefineTrait, phone: &str) -> bool {
        let vk = VddKey::UniquePhone(phone.into());
        pool.get_unique(vk).await.is_err()
    }

    pub(super) async fn u_change_name(
        pool: &impl SqlxDefineTrait,
        uid: &UserId,
        new_name: &str,
    ) -> Ares {
        let mut user = u_load(pool, uid, true, true).await;
        user.name = new_name.into();
        pool.update_entity(uid.0, &serde_json::to_string(&user).unwrap())
            .await
    }

    pub(super) async fn change_password(
        pool: &impl SqlxDefineTrait,
        uid: &UserId,
        old_pwd: &str,
        new_pwd: &str,
    ) -> Ares {
        let mut user = u_load(pool, uid, true, true).await;
        if user.password.as_ref() == old_pwd {
            user.password = new_pwd.into();
            pool.update_entity(uid.0, &serde_json::to_string(&user).unwrap())
                .await
        } else {
            Err("Passwords don't match".into())
        }
    }

    pub(super) async fn u_change_role(
        pool: &impl SqlxDefineTrait,
        target_uid: &UserId,
        is_admin: bool,
    ) -> Ares {
        let mut user = u_load(pool, &target_uid, true, true).await;
        user.role = if is_admin {
            VddUserRole::UrAdmin
        } else {
            VddUserRole::UrUser
        };
        pool.update_entity(target_uid.0, &serde_json::to_string(&user).unwrap())
            .await
    }

    pub(super) async fn u_change_key(pool: &impl SqlxDefineTrait, u_id: &UserId) -> Ares {
        let mut user = u_load(pool, u_id, true, true).await;
        user.set_key_pair();
        pool.update_entity(u_id.0, &serde_json::to_string(&user).unwrap())
            .await
    }

    pub(super) async fn u_after_login(pool: &impl SqlxDefineTrait, uid: &UserId) -> Ares {
        let mut user = u_load(pool, &uid, true, true).await;
        user.last_login_ts = Some(Local::now().naive_local());
        pool.update_entity(uid.0, &serde_json::to_string(&user).unwrap())
            .await
    }
}
